package com.kx.push.server;

import com.alibaba.fastjson.JSONObject;
import com.kx.base.util.SpringContentUtil;
import com.kx.push.mapper.MessageMapper;
import com.kx.push.pojo.Message;
import com.kx.push.pojo.MessageExample;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author Sun
 * @Date 2021/7/26 10:39
 */
@Slf4j
@Component
@ServerEndpoint("/webSocket/{uid}")
public class WebSocketServer {

    @Resource
    private MessageMapper messageMapper;

    /**
     * 统计所有和服务器连接的用户的数量
     * 基本数据类型int不行，因为此时是多线程并发
     * 所以用原子类AtomicInteger
     */
    private static AtomicInteger onlineNum = new AtomicInteger();


    /**
     * 增加一个用户
     */
    private static void addOnlineCount() {
        onlineNum.incrementAndGet();
    }

    /**
     * 减少一个用户
     */
    private static void subOnlineCount() {
        onlineNum.decrementAndGet();
    }

    /**
     * 每一个用户都和服务器建立一个会话。
     * 我们考虑保存会话
     * 一个uid对应一个session
     * 它应该是一个map
     * 用哪种map？
     * 此时线程并发场景，使用ConcurrentHashMap
     */
    private static ConcurrentHashMap<String, Session> sessionPool = new ConcurrentHashMap<>();

    /**
     * 当开启连接的时候回调该方法
     *
     * @param session 本次会话
     * @param uid     用户id
     */
    @OnOpen
    public void onOpen(Session session, @PathParam(value = "uid") String uid) {
        //刚刚连接的session放入集合中
        sessionPool.put(uid, session);
        //在线人数+1
        addOnlineCount();

        //查数据库，找出uid对应的未推送的消息，用户一上线就一次性推送给前端
        MessageExample example = new MessageExample();
        MessageExample.Criteria criteria = example.createCriteria();
        criteria.andUidEqualTo(Long.valueOf(uid));
        criteria.andStateEqualTo(0);
        //按照创建时间将数据进行排序
        example.setOrderByClause("create_time");
        //查询出符合条件的数据

        if(messageMapper==null){
            messageMapper = SpringContentUtil.getBean(MessageMapper.class);
        }

        //运行发现这里的messageMapper可能为空，所以要手动从容器中获取对象
        List<Message> messages = messageMapper.selectByExample(example);
        if(messages!=null&&messages.size()>0){
            //messages就是应该一次性推送给前端的数据
            //传输应该用json，而现在是实体类，将实体类转换成json数据
            ArrayList<JSONObject> msgs = new ArrayList<>();
            for (Message message : messages) {
                JSONObject msg = new JSONObject();
                msg.put("uid", message.getUid());
                msg.put("serviceId", message.getServiceId());
                //这里的content内容是字符串类型,别的服务器调用转化成json时会添加/,这时候我们需要将字符串转成对象
                msg.put("content",JSONObject.parseObject(message.getContent()));
                msgs.add(msg);
            }
            //整个是一个报文，完整的推送给前端
            JSONObject initInfo=new JSONObject();
            initInfo.put("init",true);
            initInfo.put("message",msgs);

            //一次性将所有数据发送给前端
            sendMessage(session,JSONObject.toJSONString(initInfo));

            //更新数据库将所有推送完成的消息状态设置成1
            Message message = new Message();
            message.setState(1);

            messageMapper.updateByExampleSelective(message, example);
        }

    }

    /**
     * 当对方发送消息时的回调
     * @param message 对方发送的消息
     */
    @OnMessage
    public void onMessage(String message){
        log.info(message);
    }

    @OnClose
    public void onClose(@PathParam(value = "uid") String uid) {
        //刚刚关闭的连接移出集合
        sessionPool.remove(uid);
        //在线人数-1
        subOnlineCount();
    }

    /**
     * 当连接发生错误时的回调
     *
     * @param session   发生错误的会话实例
     * @param throwable 发生的错误类型
     */
    @OnError
    public void onError(Session session, Throwable throwable) {
        throwable.printStackTrace();
    }

    /**
     * 根据session发送消息
     *
     * @param session 与用户的会话
     * @param message 发送给用户的内容
     */
    private void sendMessage(Session session, String message) {
        if (session != null) {
            synchronized (session) {
                try {
                    //根据一个session，发送消息内容给对方
                    session.getBasicRemote().sendText(message);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 根据用户id发送消息
     *
     * @param uid     用户id
     * @param message 需要发送消息的内容
     * @return 是否发送成功
     */
    public boolean sendInfo(String uid, String message) {
        //获取连接对象
        Session session = sessionPool.get(uid);
        if (session == null) {
            return false;
        }
        sendMessage(session, message);
        return true;
    }
}
